package com.reactim.message.service;

import com.reactim.message.dto.BatchMessageResponse;
import com.reactim.message.dto.BatchSendMessageRequest;
import com.reactim.message.dto.MessageResponse;
import com.reactim.message.dto.SendMessageRequest;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 批量消息处理服务
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class BatchMessageService {
    
    private final MessageService messageService;
    
    /**
     * 批量发送消息
     */
    public Mono<BatchMessageResponse> sendBatchMessages(BatchSendMessageRequest batchRequest) {
        long startTime = System.currentTimeMillis();
        String batchId = UUID.randomUUID().toString();
        
        log.info("开始批量发送消息: batchId={}, 消息数量={}", batchId, batchRequest.getMessages().size());
        
        // 验证批处理请求
        return validateBatchRequest(batchRequest)
                .flatMap(validatedRequest -> processBatchMessages(batchId, validatedRequest, startTime))
                .onErrorResume(error -> {
                    log.error("批量消息发送失败: batchId={}", batchId, error);
                    return createErrorResponse(batchId, batchRequest, startTime, error);
                });
    }
    
    /**
     * 验证批处理请求
     */
    private Mono<BatchSendMessageRequest> validateBatchRequest(BatchSendMessageRequest request) {
        if (request.getMessages() == null || request.getMessages().isEmpty()) {
            return Mono.error(new IllegalArgumentException("消息列表不能为空"));
        }
        
        BatchSendMessageRequest.BatchOptions options = request.getOptions();
        if (options == null) {
            options = new BatchSendMessageRequest.BatchOptions();
            request.setOptions(options);
        }
        
        // 检查批处理大小限制
        if (request.getMessages().size() > options.getBatchSize()) {
            return Mono.error(new IllegalArgumentException(
                String.format("批处理大小超过限制: %d > %d", request.getMessages().size(), options.getBatchSize())));
        }
        
        return Mono.just(request);
    }
    
    /**
     * 处理批量消息
     */
    private Mono<BatchMessageResponse> processBatchMessages(String batchId, BatchSendMessageRequest request, long startTime) {
        BatchSendMessageRequest.BatchOptions options = request.getOptions();
        
        if (options.getTransactional()) {
            // 事务模式：全部成功或全部失败
            return processTransactionalBatch(batchId, request, startTime);
        } else {
            // 非事务模式：尽力而为
            return processBestEffortBatch(batchId, request, startTime);
        }
    }
    
    /**
     * 事务模式批处理
     */
    private Mono<BatchMessageResponse> processTransactionalBatch(String batchId, BatchSendMessageRequest request, long startTime) {
        return Flux.fromIterable(request.getMessages())
                .index()
                .flatMap(indexedMessage -> {
                    int index = indexedMessage.getT1().intValue();
                    SendMessageRequest messageRequest = indexedMessage.getT2();
                    
                    return messageService.sendMessage(messageRequest)
                            .map(response -> new IndexedResult(index, response, null))
                            .onErrorResume(error -> Mono.just(new IndexedResult(index, null, error)));
                })
                .collectList()
                .flatMap(results -> {
                    // 检查是否有失败的消息
                    boolean hasFailures = results.stream().anyMatch(result -> result.error != null);
                    
                    if (hasFailures) {
                        // 有失败的消息，回滚所有操作（这里简化处理）
                        log.warn("事务批处理中有失败消息，回滚操作: batchId={}", batchId);
                        return Mono.just(createFailureResponse(batchId, request, results, startTime));
                    } else {
                        // 全部成功
                        return Mono.just(createSuccessResponse(batchId, request, results, startTime));
                    }
                });
    }
    
    /**
     * 尽力而为模式批处理
     */
    private Mono<BatchMessageResponse> processBestEffortBatch(String batchId, BatchSendMessageRequest request, long startTime) {
        BatchSendMessageRequest.BatchOptions options = request.getOptions();
        
        Flux<IndexedResult> resultFlux = Flux.fromIterable(request.getMessages())
                .index()
                .flatMap(indexedMessage -> {
                    int index = indexedMessage.getT1().intValue();
                    SendMessageRequest messageRequest = indexedMessage.getT2();
                    
                    return processMessageWithRetry(messageRequest, options.getRetryCount())
                            .map(response -> new IndexedResult(index, response, null))
                            .onErrorResume(error -> Mono.just(new IndexedResult(index, null, error)));
                }, options.getParallel() ? 10 : 1); // 并行度控制
        
        if (options.getParallel()) {
            resultFlux = resultFlux.subscribeOn(Schedulers.parallel());
        }
        
        return resultFlux
                .timeout(Duration.ofMillis(options.getTimeoutMs()))
                .collectList()
                .map(results -> createMixedResponse(batchId, request, results, startTime));
    }
    
    /**
     * 带重试的消息处理
     */
    private Mono<MessageResponse> processMessageWithRetry(SendMessageRequest request, int retryCount) {
        return messageService.sendMessage(request)
                .retry(retryCount)
                .doOnError(error -> log.warn("消息发送失败，已重试{}次: {}", retryCount, error.getMessage()));
    }
    
    /**
     * 创建成功响应
     */
    private BatchMessageResponse createSuccessResponse(String batchId, BatchSendMessageRequest request, 
                                                      List<IndexedResult> results, long startTime) {
        List<MessageResponse> successMessages = new ArrayList<>();
        
        for (IndexedResult result : results) {
            if (result.response != null) {
                successMessages.add(result.response);
            }
        }
        
        return BatchMessageResponse.builder()
                .batchId(batchId)
                .totalCount(request.getMessages().size())
                .successCount(successMessages.size())
                .failureCount(0)
                .successMessages(successMessages)
                .failedMessages(new ArrayList<>())
                .processingTimeMs(System.currentTimeMillis() - startTime)
                .status(BatchMessageResponse.BatchStatus.ALL_SUCCESS)
                .build();
    }
    
    /**
     * 创建失败响应
     */
    private BatchMessageResponse createFailureResponse(String batchId, BatchSendMessageRequest request, 
                                                      List<IndexedResult> results, long startTime) {
        List<BatchMessageResponse.FailedMessage> failedMessages = new ArrayList<>();
        
        for (IndexedResult result : results) {
            if (result.error != null) {
                failedMessages.add(BatchMessageResponse.FailedMessage.builder()
                        .requestIndex(result.index)
                        .errorMessage(result.error.getMessage())
                        .errorCode("SEND_FAILED")
                        .originalRequest(request.getMessages().get(result.index))
                        .build());
            }
        }
        
        return BatchMessageResponse.builder()
                .batchId(batchId)
                .totalCount(request.getMessages().size())
                .successCount(0)
                .failureCount(request.getMessages().size())
                .successMessages(new ArrayList<>())
                .failedMessages(failedMessages)
                .processingTimeMs(System.currentTimeMillis() - startTime)
                .status(BatchMessageResponse.BatchStatus.ALL_FAILED)
                .build();
    }
    
    /**
     * 创建混合响应（部分成功部分失败）
     */
    private BatchMessageResponse createMixedResponse(String batchId, BatchSendMessageRequest request, 
                                                    List<IndexedResult> results, long startTime) {
        List<MessageResponse> successMessages = new ArrayList<>();
        List<BatchMessageResponse.FailedMessage> failedMessages = new ArrayList<>();
        
        for (IndexedResult result : results) {
            if (result.response != null) {
                successMessages.add(result.response);
            } else if (result.error != null) {
                failedMessages.add(BatchMessageResponse.FailedMessage.builder()
                        .requestIndex(result.index)
                        .errorMessage(result.error.getMessage())
                        .errorCode("SEND_FAILED")
                        .originalRequest(request.getMessages().get(result.index))
                        .build());
            }
        }
        
        BatchMessageResponse.BatchStatus status;
        if (successMessages.size() == request.getMessages().size()) {
            status = BatchMessageResponse.BatchStatus.ALL_SUCCESS;
        } else if (successMessages.isEmpty()) {
            status = BatchMessageResponse.BatchStatus.ALL_FAILED;
        } else {
            status = BatchMessageResponse.BatchStatus.PARTIAL_SUCCESS;
        }
        
        return BatchMessageResponse.builder()
                .batchId(batchId)
                .totalCount(request.getMessages().size())
                .successCount(successMessages.size())
                .failureCount(failedMessages.size())
                .successMessages(successMessages)
                .failedMessages(failedMessages)
                .processingTimeMs(System.currentTimeMillis() - startTime)
                .status(status)
                .build();
    }
    
    /**
     * 创建错误响应
     */
    private Mono<BatchMessageResponse> createErrorResponse(String batchId, BatchSendMessageRequest request, 
                                                          long startTime, Throwable error) {
        return Mono.just(BatchMessageResponse.builder()
                .batchId(batchId)
                .totalCount(request.getMessages().size())
                .successCount(0)
                .failureCount(request.getMessages().size())
                .successMessages(new ArrayList<>())
                .failedMessages(List.of(BatchMessageResponse.FailedMessage.builder()
                        .requestIndex(-1)
                        .errorMessage(error.getMessage())
                        .errorCode("BATCH_PROCESSING_ERROR")
                        .build()))
                .processingTimeMs(System.currentTimeMillis() - startTime)
                .status(BatchMessageResponse.BatchStatus.ALL_FAILED)
                .build());
    }
    
    /**
     * 索引结果内部类
     */
    private static class IndexedResult {
        final int index;
        final MessageResponse response;
        final Throwable error;
        
        IndexedResult(int index, MessageResponse response, Throwable error) {
            this.index = index;
            this.response = response;
            this.error = error;
        }
    }
}